package com.shujia.kafka

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import scala.io.Source

object Demo2StudentToKafka {
  def main(args: Array[String]): Unit = {

    /**
      * 1、创佳生产者
      *
      */

    val properties = new Properties

    //1、指定kafkabroker地址
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")

    //2、指定kv的序列化类
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")


    val producer = new KafkaProducer[String, String](properties)


    //读取学生表
    val students: Iterator[String] = Source.fromFile("data/students.txt").getLines()


    //    kafka-topics.sh --create --zookeeper master:2181,node1:2181,node2:2181 --replication-factor 2 --partitions 2 --topic student2


    //循环一条一条将数据写入kafka
    students.foreach(line => {
      val id: String = line.split(",")(0)

      //将不同的性别写入不同分区
      val gender: String = line.split(",")(3)


      //手动指定分区
      val partition: Int = gender match {
        case "男" => 1
        case _ => 0
      }

      val record = new ProducerRecord[String, String]("student3", partition, id, line)

      producer.send(record)

      producer.flush()
    })

    producer.close()


  }

}
